/*
    * Copyright 2002-2019 the original author or authors.
    *
    * Licensed under the Apache License, Version 2.0 (the "License");
    * you may not use this file except in compliance with the License.
    * You may obtain a copy of the License at
    *
    *      https://www.apache.org/licenses/LICENSE-2.0
    *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  
  package org.springframework.integration.jpa.core;
  
  import java.util.List;
  
  import javax.persistence.EntityManager;
  import javax.persistence.EntityManagerFactory;
  
  import org.springframework.beans.BeansException;
  import org.springframework.beans.factory.BeanFactory;
  import org.springframework.beans.factory.BeanFactoryAware;
  import org.springframework.beans.factory.InitializingBean;
  import org.springframework.expression.EvaluationContext;
  import org.springframework.expression.Expression;
  import org.springframework.expression.common.LiteralExpression;
  import org.springframework.integration.expression.ExpressionUtils;
  import org.springframework.integration.jpa.support.JpaParameter;
  import org.springframework.integration.jpa.support.PersistMode;
  import org.springframework.integration.jpa.support.parametersource.BeanPropertyParameterSourceFactory;
  import org.springframework.integration.jpa.support.parametersource.ExpressionEvaluatingParameterSourceFactory;
  import org.springframework.integration.jpa.support.parametersource.ParameterSource;
  import org.springframework.integration.jpa.support.parametersource.ParameterSourceFactory;
  import org.springframework.lang.Nullable;
  import org.springframework.messaging.Message;
  import org.springframework.messaging.MessagingException;
  import org.springframework.util.Assert;
  import org.springframework.util.CollectionUtils;
  
  /**
   * Executes Jpa Operations that produce payload objects from the result of the provided:
   *
   * <ul>
   *     <li>entityClass</li>
   *     <li>JpQl Select Query</li>
   *     <li>Sql Native Query</li>
   *     <li>JpQl Named Query</li>
   *     <li>Sql Native Named Query</li>
   * </ul>
   *
   * When objects are being retrieved, it also possibly to:
   *
   * <ul>
   *     <li>delete the retrieved object</li>
   * </ul>
   *
   * If neither entityClass nor any other query is specified then the entity-class
   * is "guessed" from the {@link Message} payload.idExpression
   *
   * @author Gunnar Hillert
   * @author Amol Nayak
   * @author Artem Bilan
   *
   * @since 2.2
   *
   */
  public class JpaExecutor implements InitializingBean, BeanFactoryAware {
  
  	private final JpaOperations jpaOperations;
  
  	private List<JpaParameter> jpaParameters;
  
  	private Class<?> entityClass;
  
  	private String jpaQuery;
  
  	private String nativeQuery;
  
  	private String namedQuery;
  
  	private Expression maxResultsExpression;
  
  	private Expression firstResultExpression;
  
  	private Expression idExpression;
  
  	private PersistMode persistMode = PersistMode.MERGE;
  
  	private ParameterSourceFactory parameterSourceFactory = null;
  
  	private ParameterSource parameterSource;
  
  	private boolean flush = false;
  
  	private int flushSize = 0;
 
 	private boolean clearOnFlush = false;
 
 	private boolean deleteAfterPoll = false;
 
 	private boolean deleteInBatch = false;
 
 	private boolean expectSingleResult = false;
 
 	/**
 	 * Indicates that whether only the payload of the passed in {@link Message}
 	 * will be used as a source of parameters. The is 'true' by default because as a
 	 * default a {@link BeanPropertyParameterSourceFactory} implementation is
 	 * used for the sqlParameterSourceFactory property.
 	 */
 	private Boolean usePayloadAsParameterSource = null;
 
 	private BeanFactory beanFactory;
 
 	private EvaluationContext evaluationContext;
 
 	/**
 	 * Constructor taking an {@link EntityManagerFactory} from which the
 	 * {@link EntityManager} can be obtained.
 	 * @param entityManagerFactory Must not be null.
 	 */
 	public JpaExecutor(EntityManagerFactory entityManagerFactory) {
 		Assert.notNull(entityManagerFactory, "entityManagerFactory must not be null.");
 
 		DefaultJpaOperations defaultJpaOperations = new DefaultJpaOperations();
 		defaultJpaOperations.setEntityManagerFactory(entityManagerFactory);
 		defaultJpaOperations.afterPropertiesSet();
 
 		this.jpaOperations = defaultJpaOperations;
 	}
 
 	/**
 	 * Constructor taking an {@link EntityManager} directly.
 	 * @param entityManager Must not be null.
 	 */
 	public JpaExecutor(EntityManager entityManager) {
 		Assert.notNull(entityManager, "entityManager must not be null.");
 
 		DefaultJpaOperations defaultJpaOperations = new DefaultJpaOperations();
 		defaultJpaOperations.setEntityManager(entityManager);
 		defaultJpaOperations.afterPropertiesSet();
 		this.jpaOperations = defaultJpaOperations;
 	}
 
 	/**
 	 * If custom behavior is required a custom implementation of {@link JpaOperations}
 	 * can be passed in. The implementations themselves typically provide access
 	 * to the {@link EntityManager}.
 	 * See also {@link DefaultJpaOperations} and {@link AbstractJpaOperations}.
 	 * @param jpaOperations Must not be null.
 	 */
 	public JpaExecutor(JpaOperations jpaOperations) {
 		Assert.notNull(jpaOperations, "jpaOperations must not be null.");
 		this.jpaOperations = jpaOperations;
 	}
 
 	public void setIntegrationEvaluationContext(EvaluationContext evaluationContext) {
 		this.evaluationContext = evaluationContext;
 	}
 
 	/**
 	 * Sets the class type which is being used for retrieving entities from the
 	 * database.
 	 * @param entityClass Must not be null.
 	 */
 	public void setEntityClass(Class<?> entityClass) {
 		Assert.notNull(entityClass, "entityClass must not be null.");
 		this.entityClass = entityClass;
 	}
 
 	/**
 	 * @param jpaQuery The provided JPA query must neither be null nor empty.
 	 */
 	public void setJpaQuery(String jpaQuery) {
 		Assert.isTrue(this.nativeQuery == null && this.namedQuery == null, "You can define only one of the "
 				+ "properties 'jpaQuery', 'nativeQuery', 'namedQuery'");
 		Assert.hasText(jpaQuery, "jpaQuery must neither be null nor empty.");
 		this.jpaQuery = jpaQuery;
 	}
 
 	/**
 	 * You can also use native Sql queries to poll data from the database. If set
 	 * this property will allow you to use native SQL. Optionally you can also set
 	 * the entityClass property at the same time. If specified the entityClass will
 	 * be used as the result class for the native query.
 	 * @param nativeQuery The provided SQL query must neither be null nor empty.
 	 */
 	public void setNativeQuery(String nativeQuery) {
 
 		Assert.isTrue(this.namedQuery == null && this.jpaQuery == null, "You can define only one of the "
 				+ "properties 'jpaQuery', 'nativeQuery', 'namedQuery'");
 		Assert.hasText(nativeQuery, "nativeQuery must neither be null nor empty.");
 
 		this.nativeQuery = nativeQuery;
 	}
 
 	/**
 	 * A named query can either refer to a named JPQL based query or a native SQL
 	 * query.
 	 * @param namedQuery Must neither be null nor empty
 	 */
 	public void setNamedQuery(String namedQuery) {
 
 		Assert.isTrue(this.jpaQuery == null && this.nativeQuery == null, "You can define only one of the "
 				+ "properties 'jpaQuery', 'nativeQuery', 'namedQuery'");
 
 		Assert.hasText(namedQuery, "namedQuery must neither be null nor empty.");
 		this.namedQuery = namedQuery;
 	}
 
 	public void setPersistMode(PersistMode persistMode) {
 		this.persistMode = persistMode;
 	}
 
 	public void setJpaParameters(List<JpaParameter> jpaParameters) {
 		this.jpaParameters = jpaParameters;
 	}
 
 	public void setUsePayloadAsParameterSource(Boolean usePayloadAsParameterSource) {
 		this.usePayloadAsParameterSource = usePayloadAsParameterSource;
 	}
 
 	/**
 	 * If set to {@code true} the {@link javax.persistence.EntityManager#flush()} will be called
 	 * after persistence operation.
 	 * Has the same effect, if the {@link #flushSize} is specified to {@code 1}.
 	 * For convenience in cases when the provided entity to persist is not an instance of {@link Iterable}.
 	 * @param flush defaults to 'false'.
 	 */
 	public void setFlush(boolean flush) {
 		this.flush = flush;
 	}
 
 	/**
 	 * If the provided value is greater than {@code 0}, then {@link javax.persistence.EntityManager#flush()}
 	 * will be called after persistence operations as well as within batch operations.
 	 * This property has precedence over the {@link #flush}, if it is specified to a value greater than {@code 0}.
 	 * If the entity to persist is not an instance of {@link Iterable} and this property is greater than {@code 0},
 	 * then the entity will be flushed as if the {@link #flush} attribute was set to {@code true}.
 	 * @param flushSize defaults to '0'.
 	 */
 	public void setFlushSize(int flushSize) {
 		Assert.state(flushSize >= 0, "'flushSize' cannot be less than '0'.");
 		this.flushSize = flushSize;
 	}
 
 	/**
 	 * If set to {@code true} the {@link javax.persistence.EntityManager#clear()} will be called,
 	 * and only if the {@link javax.persistence.EntityManager#flush()} was called after performing persistence
 	 * operations.
 	 * @param clearOnFlush defaults to 'false'.
 	 * @see #setFlush(boolean)
 	 * @see #setFlushSize(int)
 	 */
 	public void setClearOnFlush(boolean clearOnFlush) {
 		this.clearOnFlush = clearOnFlush;
 	}
 
 	/**
 	 * If not set, this property defaults to <code>false</code>, which means that
 	 * deletion occurs on a per object basis if a collection of entities is being
 	 * deleted.
 	 *<p>If set to 'true' the elements of the payload are deleted as a batch
 	 * operation. Be aware that this exhibits issues in regards to cascaded deletes.
 	 *<p>The specification 'JSR 317: Java Persistence API, Version 2.0' does not
 	 * support cascaded deletes in batch operations. The specification states in
 	 * chapter 4.10:
 	 *<p>"A delete operation only applies to entities of the specified class and
 	 * its subclasses. It does not cascade to related entities."
 	 * @param deleteInBatch Defaults to 'false' if not set.
 	 */
 	public void setDeleteInBatch(boolean deleteInBatch) {
 		this.deleteInBatch = deleteInBatch;
 	}
 
 	/**
 	 * If set to 'true', the retrieved objects are deleted from the database upon
 	 * being polled. May not work in all situations, e.g. for Native SQL Queries.
 	 * @param deleteAfterPoll Defaults to 'false'.
 	 */
 	public void setDeleteAfterPoll(boolean deleteAfterPoll) {
 		this.deleteAfterPoll = deleteAfterPoll;
 	}
 
 	/**
 	 * @param parameterSourceFactory Must not be null
 	 */
 	public void setParameterSourceFactory(ParameterSourceFactory parameterSourceFactory) {
 		Assert.notNull(parameterSourceFactory, "parameterSourceFactory must not be null.");
 		this.parameterSourceFactory = parameterSourceFactory;
 	}
 
 	/**
 	 * Specify the {@link ParameterSource} that would be used to provide
 	 * additional parameters.
 	 * @param parameterSource Must not be null.
 	 */
 	public void setParameterSource(ParameterSource parameterSource) {
 		Assert.notNull(parameterSource, "parameterSource must not be null.");
 		this.parameterSource = parameterSource;
 	}
 
 	/**
 	 *
 	 * This parameter indicates that only one result object shall be returned as
 	 * a result from the executed JPA operation. If set to <code>true</code> and
 	 * the result list from the JPA operations contains only 1 element, then that
 	 * 1 element is extracted and returned as payload.
 	 * <p>If the result map contains more than 1 element and
 	 * {@link JpaExecutor#expectSingleResult} is <code>true</code>, then a
 	 * {@link MessagingException} is thrown.
 	 * <p>If set to <code>false</code>, the complete result list is returned as the
 	 * payload.
 	 * @param expectSingleResult true if a single object is expected.
 	 *
 	 */
 	public void setExpectSingleResult(boolean expectSingleResult) {
 		this.expectSingleResult = expectSingleResult;
 	}
 
 	/**
 	 * Set the expression that will be evaluated to get the first result in the query executed.
 	 * If a null expression is set, all the results in the result set will be retrieved
 	 * @param firstResultExpression The first result expression.
 	 * @see javax.persistence.Query#setFirstResult(int)
 	 */
 	public void setFirstResultExpression(Expression firstResultExpression) {
 		this.firstResultExpression = firstResultExpression;
 	}
 
 	/**
 	 * Set the expression that will be evaluated to get the {@code primaryKey} for
 	 * {@link javax.persistence.EntityManager#find(Class, Object)}
 	 * @param idExpression the SpEL expression for entity {@code primaryKey}.
 	 * @since 4.0
 	 */
 	public void setIdExpression(Expression idExpression) {
 		this.idExpression = idExpression;
 	}
 
 	/**
 	 * Set the expression for maximum number of results expression. It has be a non null value
 	 * Not setting one will default to the behavior of fetching all the records
 	 * @param maxResultsExpression The maximum results expression.
 	 */
 	public void setMaxResultsExpression(Expression maxResultsExpression) {
 		Assert.notNull(maxResultsExpression, "maxResultsExpression cannot be null");
 		this.maxResultsExpression = maxResultsExpression;
 	}
 
 	/**
 	 * Set the max number of results to retrieve from the database. Defaults to
 	 * 0, which means that all possible objects shall be retrieved.
 	 * @param maxNumberOfResults Must not be negative.
 	 * @see javax.persistence.Query#setMaxResults(int)
 	 */
 	public void setMaxNumberOfResults(int maxNumberOfResults) {
 		this.setMaxResultsExpression(new LiteralExpression("" + maxNumberOfResults));
 	}
 
 	@Override
 	public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
 		this.beanFactory = beanFactory;
 	}
 
 	/**
 	 * Verify and sets the parameters. E.g. initializes the to be used
 	 * {@link ParameterSourceFactory}.
 	 */
 	@Override
 	public void afterPropertiesSet() {
 		if (!CollectionUtils.isEmpty(this.jpaParameters)) {
 			if (this.parameterSourceFactory == null) {
 				ExpressionEvaluatingParameterSourceFactory expressionSourceFactory =
 						new ExpressionEvaluatingParameterSourceFactory(this.beanFactory);
 				expressionSourceFactory.setParameters(this.jpaParameters);
 				this.parameterSourceFactory = expressionSourceFactory;
 
 			}
 			else {
 				throw new IllegalStateException("The 'jpaParameters' and 'parameterSourceFactory' " +
 						"are mutually exclusive. Consider to configure parameters on the provided " +
 						"'parameterSourceFactory': " + this.parameterSourceFactory);
 			}
 
 			if (this.usePayloadAsParameterSource == null) {
 				this.usePayloadAsParameterSource = false;
 			}
 
 		}
 		else {
 
 			if (this.parameterSourceFactory == null) {
 				this.parameterSourceFactory = new BeanPropertyParameterSourceFactory();
 			}
 
 			if (this.usePayloadAsParameterSource == null) {
 				this.usePayloadAsParameterSource = true;
 			}
 		}
 
 		if (this.flushSize > 0) {
 			this.flush = true;
 		}
 		else if (this.flush) {
 			this.flushSize = 1;
 		}
 
 		if (this.evaluationContext == null) {
 			this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(this.beanFactory);
 		}
 	}
 
 	/**
 	 * Execute the actual Jpa Operation. Call this method, if you need access to
 	 * process return values. This methods return a Map that contains either
 	 * the number of affected entities or the affected entity itself.
 	 *<p>Keep in mind that the number of entities effected by the operation may
 	 * not necessarily correlate with the number of rows effected in the database.
 	 * @param message The message.
 	 * @return Either the number of affected entities when using a JPQL query.
 	 * When using a merge/persist the updated/inserted itself is returned.
 	 */
 	public Object executeOutboundJpaOperation(Message<?> message) {
 		ParameterSource paramSource = null;
 		if (this.jpaQuery != null || this.nativeQuery != null || this.namedQuery != null) {
 			paramSource = determineParameterSource(message);
 		}
 		if (this.jpaQuery != null) {
 			return this.jpaOperations.executeUpdate(this.jpaQuery, paramSource);
 		}
 		else if (this.nativeQuery != null) {
 			return this.jpaOperations.executeUpdateWithNativeQuery(this.nativeQuery, paramSource);
 		}
 		else if (this.namedQuery != null) {
 			return this.jpaOperations.executeUpdateWithNamedQuery(this.namedQuery, paramSource);
 		}
 		else {
 			return executeOutboundJpaOperationOnPersistentMode(message);
 		}
 	}
 
 	private Object executeOutboundJpaOperationOnPersistentMode(Message<?> message) {
 		Object payload = message.getPayload();
 		switch (this.persistMode) {
 			case PERSIST:
 				this.jpaOperations.persist(payload, this.flushSize, this.clearOnFlush);
 				return payload;
 			case MERGE:
 				return this.jpaOperations.merge(payload, this.flushSize, this.clearOnFlush); // NOSONAR
 			case DELETE:
 				this.jpaOperations.delete(payload);
 				if (this.flush) {
 					this.jpaOperations.flush();
 				}
 				return payload;
 			default:
 				throw new IllegalStateException("Unsupported PersistMode: " + this.persistMode.name());
 		}
 	}
 
 	/**
 	 * Execute the JPA operation. Delegates to {@link JpaExecutor#poll(Message)}.
 	 * @return The object or null.
 	 */
 	@Nullable
 	public Object poll() {
 		return poll(null);
 	}
 
 	/**
 	 * Execute a (typically retrieving) JPA operation. The <i>requestMessage</i>
 	 * can be used to provide additional query parameters using
 	 * {@link JpaExecutor#parameterSourceFactory}. If the
 	 * <i>requestMessage</i> parameter is null then
 	 * {@link JpaExecutor#parameterSource} is being used for providing query parameters.
 	 * @param requestMessage May be null.
 	 * @return The payload object, which may be null.
 	 */
 	@Nullable
 	public Object poll(@Nullable final Message<?> requestMessage) {
 		final Object payload;
 
 		if (this.idExpression != null) {
 			Object id = this.idExpression.getValue(this.evaluationContext, requestMessage); // NOSONAR It can be null
 			Assert.state(id != null, "The 'idExpression' cannot evaluate to null.");
 			Class<?> entityClazz = this.entityClass;
 			if (entityClazz == null && requestMessage != null) {
 				entityClazz = requestMessage.getPayload().getClass();
 			}
 			Assert.state(entityClazz != null, "The entity class to retrieve cannot be null.");
 			payload = this.jpaOperations.find(entityClazz, id);
 		}
 		else {
 			final List<?> result;
 			int maxNumberOfResults = evaluateExpressionForNumericResult(requestMessage, this.maxResultsExpression);
 			if (requestMessage == null) {
 				result = doPoll(this.parameterSource, 0, maxNumberOfResults);
 			}
 			else {
 				int firstResult = 0;
 				if (this.firstResultExpression != null) {
 					firstResult = getFirstResult(requestMessage);
 				}
 				ParameterSource paramSource = determineParameterSource(requestMessage);
 				result = doPoll(paramSource, firstResult, maxNumberOfResults);
 			}
 
 			if (result.isEmpty()) {
 				payload = null;
 			}
 			else {
 				if (this.expectSingleResult) {
 					if (result.size() == 1) {
 						payload = result.iterator().next();
 					}
 					else if (requestMessage != null) {
 						throw new MessagingException(requestMessage,
 								"The Jpa operation returned more than 1 result for expectSingleResult mode.");
 					}
 					else {
 						throw new MessagingException(
 								"The Jpa operation returned more than 1 result for expectSingleResult mode.");
 					}
 				}
 				else {
 					payload = result;
 				}
 			}
 		}
 
 		checkDelete(payload);
 		return payload;
 	}
 
 	private void checkDelete(final Object payload) {
 		if (payload != null && this.deleteAfterPoll) {
 			if (payload instanceof Iterable) {
 				if (this.deleteInBatch) {
 					this.jpaOperations.deleteInBatch((Iterable<?>) payload);
 				}
 				else {
 					for (Object entity : (Iterable<?>) payload) {
 						this.jpaOperations.delete(entity);
 					}
 				}
 			}
 			else {
 				this.jpaOperations.delete(payload);
 			}
 
 			if (this.flush) {
 				this.jpaOperations.flush();
 			}
 		}
 	}
 
 	protected List<?> doPoll(ParameterSource jpaQLParameterSource, int firstResult, int maxNumberOfResults) {
 		List<?> payload;
 		if (this.jpaQuery != null) {
 			payload =
 					this.jpaOperations.getResultListForQuery(this.jpaQuery, jpaQLParameterSource,
 							firstResult, maxNumberOfResults);
 		}
 		else if (this.nativeQuery != null) {
 			payload =
 					this.jpaOperations.getResultListForNativeQuery(this.nativeQuery, this.entityClass,
 							jpaQLParameterSource, firstResult, maxNumberOfResults);
 		}
 		else if (this.namedQuery != null) {
 			payload =
 					this.jpaOperations.getResultListForNamedQuery(this.namedQuery, jpaQLParameterSource,
 							firstResult, maxNumberOfResults);
 		}
 		else if (this.entityClass != null) {
 			payload = this.jpaOperations.getResultListForClass(this.entityClass, firstResult, maxNumberOfResults);
 		}
 		else {
 			throw new IllegalStateException("For the polling operation, one of "
 					+ "the following properties must be specified: "
 					+ "query, namedQuery or entityClass.");
 		}
 		return payload;
 	}
 
 	private int getFirstResult(final Message<?> requestMessage) {
 		return evaluateExpressionForNumericResult(requestMessage, this.firstResultExpression);
 	}
 
 	private int evaluateExpressionForNumericResult(@Nullable final Message<?> requestMessage,
 			@Nullable Expression expression) {
 
 		int evaluatedResult = 0;
 		if (expression != null) {
 			Object evaluationResult = expression.getValue(this.evaluationContext, requestMessage); // NOSONAR can be
 			// null
 			if (evaluationResult != null) {
 				if (evaluationResult instanceof Number) {
 					evaluatedResult = ((Number) evaluationResult).intValue();
 				}
 				else if (evaluationResult instanceof String) {
 					try {
 						evaluatedResult = Integer.parseInt((String) evaluationResult);
 					}
 					catch (NumberFormatException e) {
 						throw new IllegalArgumentException(
 								"Value " + evaluationResult + " passed as cannot be " +
 										"parsed to a number, expected to be numeric", e);
 					}
 				}
 				else {
 					throw new IllegalArgumentException("Expected the value to be a Number got "
 							+ evaluationResult.getClass().getName());
 				}
 			}
 		}
 		return evaluatedResult;
 	}
 
 	private ParameterSource determineParameterSource(final Message<?> requestMessage) {
 		if (this.usePayloadAsParameterSource) {
 			return this.parameterSourceFactory.createParameterSource(requestMessage.getPayload());
 		}
 		else {
 			return this.parameterSourceFactory.createParameterSource(requestMessage);
 		}
 	}
 
 }